package com.demo.rpc.manager;

import com.demo.rpc.exception.RPCException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * Created by hunter
 *
 * @Description 具体有2pc/3pc/tcc解决方案, 这里利用redis实现一个简单的自我实现的方案
 */
@Component
public class RPCTransactionalManager {

    // 最长事务等待时间, 暂时给这么多
    private static final Long MAX_WAIT_TIME = 10 * 1000L;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    /**
     * 当前应用名
     */
    @Value("${spring.application.name:null}")
    private String applicationName;

    /**
     * 是否是当前应用开启的分布式事务, 这里单个应用多进程时需要修改适配, 先简单处理 TODO
     *
     * @param transactionId
     * @return
     */
    public boolean isCurrentApplicationStart(String transactionId) {
        if (transactionId == null) {
            return false;
        }

        return transactionId.startsWith(applicationName);
    }

    /**
     * 创建一个分布式事务, 这里单个应用多进程时需要修改适配, 先简单处理 TODO
     *
     * @return
     */
    public String createRPCTransactional() {
        String transactionId = applicationName + "-" + UUID.randomUUID().toString();
        redisTemplate.opsForValue().set(transactionId, "0");
        redisTemplate.expire(transactionId, MAX_WAIT_TIME + 5000, TimeUnit.MILLISECONDS);
        return transactionId;
    }

    /**
     * 开始分布式事务任务
     *
     * @param transactionId
     */
    public void startRPCTransactionalTask(String transactionId) {
        if (transactionId == null) {
            return;
        }
        redisTemplate.opsForValue().increment(transactionId, 1L);
    }

    /**
     * 等待分布式事务任务完成
     *
     * @param transactionId
     */
    public void waitRPCTransactionalTask(String transactionId) {
        if (transactionId == null) {
            return;
        }
        // 初始时间
        long beginTime = System.currentTimeMillis();
        // 等待所有事务完成
        while (true) {
            // 拿到任务
            String transactionCurrentCountStr = redisTemplate.opsForValue().get(transactionId);
            // 任务为空不处理
            if (transactionCurrentCountStr == null) {
                return;
            }
            Long transactionCurrentCount = Long.valueOf(transactionCurrentCountStr);

            // 所有的事务任务已经执行完毕
            if (transactionCurrentCount == 0) {
                return;
            } else if (transactionCurrentCount < 0) { // 为负数表示某一个事务出错了
                throw new RPCException("~------- RPCTransactional was occur error and interrupted ...");
            }

            long waitTime = System.currentTimeMillis();
            if (waitTime - beginTime > MAX_WAIT_TIME) {
                // 方法异常了要打断当前分布式事务
                interruptedRPCTransactionalTask(transactionId);
                throw new RPCException("~------- RPCTransactional wait timeout ...");
            }

            // 沉睡0.5秒, 别读取redis过于频繁
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 完成本地分布式事务的任务
     *
     * @param transactionId
     */
    public void completeRPCTransactionalTask(String transactionId) {
        if (transactionId == null) {
            return;
        }
        redisTemplate.opsForValue().increment(transactionId, - 1L);
    }

    /**
     * 打断当前分布式事务
     *
     * @param transactionId
     */
    public void interruptedRPCTransactionalTask(String transactionId) {
        if (transactionId == null) {
            return;
        }
        redisTemplate.opsForValue().set(transactionId, "-1");
    }
}
